#pragma once
#include <functional>
#include <unordered_map>
#include <set>
#include <chrono>
#include "tcpserver.h"
#include "udpserver.h"
#ifdef __linux__
#include "localstreamserver.h"
#include "localdgramserver.h"
#endif
#include "packet.h"
#include "server.h"

namespace KRPC{

struct SessionData
{
	uint32_t id;
	std::chrono::steady_clock::time_point tp;
	bool expire(int maxMsec = 1000)
	{
		auto now = std::chrono::steady_clock::now();
		if (tp != std::chrono::steady_clock::time_point::min()) {
			return std::chrono::duration_cast<std::chrono::milliseconds>(now - tp).count() >= maxMsec;
		} else {
			tp = now;
			return false;
		}
	}
};

enum class ServerType {
	TCP,
	UDP,
	LOCALSTREAM,
	LOCALDGRAM
};

template<ServerType TYPE = ServerType::TCP>
struct ServerKind {
	using type = common::TcpServer<MsgHead, SessionData>;
	using start_arg_type = uint16_t;
};
template<>
struct ServerKind<ServerType::UDP> {
	using type = common::UdpServer<MsgHead, SessionData>;
	using start_arg_type = uint16_t;
};
#ifdef __linux__
template<>
struct ServerKind<ServerType::LOCALDGRAM> {
	using type = common::LocalDgramServer<MsgHead, SessionData>;
	using start_arg_type = const std::string;
};
template<>
struct ServerKind<ServerType::LOCALSTREAM> {
	using type = common::LocalStreamServer<MsgHead, SessionData>;
	using start_arg_type = const std::string;
};
#endif

template<ServerType TYPE>
class NetServer : public ServerKind<TYPE>::type, public Server
{
public:
	using AccessCallback = std::function<bool(MsgHead *, uint32_t)>;
	using Base = typename ServerKind<TYPE>::type;
	//其中_arg的类型对于udp和tcp服务是uint16_t即端口号,对于unix域stream或者dgram服务则类型为const std::string即路径字符串,且此时access会默认赋值为nullptr即不进行权限检查
	NetServer(typename ServerKind<TYPE>::start_arg_type _arg, AccessCallback callback = nullptr) : arg(_arg), access((TYPE == ServerType::LOCALSTREAM || TYPE == ServerType::LOCALDGRAM) ? nullptr : std::move(callback)) {}

	virtual void run(std::function<void(Packet &)> callback, SubcribeCallback subCallback) 
	{
		if (!callback)
			return;
		if (subCallback)
			onSubcribe = std::move(subCallback);
		onRequest = std::move(callback);
		Base::start(arg);
	}

	virtual void publish(Packet &&packet) 
	{
		std::shared_ptr<KRPC::Buffer> buf = std::make_shared<KRPC::Buffer>(packet.release());
		auto task = [buf, this]{
			MsgHead *head = static_cast<MsgHead *>(buf->data());
			auto it = subs.find(head->name);
			if (it != subs.end()) {
				for (auto iit = it->second.begin(); iit != it->second.end();) {
					auto sptr = iit->lock();
					if (sptr && !sptr->userdata.expire()) {	
						head->id = sptr->userdata.id;
						if (++iit == it->second.end()) {	//最后一个订阅者直接使用buf进行发布
							sptr->add_sndmsg(std::move(buf));
						} else {	//由于可能有多个订阅者,所以需要复制packet数据之后再进行发送
							auto tmp = std::make_shared<KRPC::Buffer>(buf->len());
							tmp->write((const char *)buf->data(), buf->len());
							sptr->add_sndmsg(std::move(tmp));
						}
					} else {
						it->second.erase(iit++);
					}
				}
			}
		};
		Base::post(task);
	}

	virtual int onHead(MsgHead &head) {return head.len - sizeof(MsgHead);}

	virtual bool onMessage(std::shared_ptr<KRPC::Buffer> buffer, typename Base::SessionSPtr pSession)
	{
		MsgHead *head = (MsgHead *)buffer->data();
		if (access && !access(head, inet_addr(pSession->address().c_str()))) {
			head->len = sizeof(MsgHead);		//当权限校验不通过时access会把head->mode修改为RPC_NO_ACCESS或者RPC_NOT_FOUND
			buffer->reset(head->len);
			pSession->add_sndmsg(std::move(buffer));
			return true;
		}
		bool ret = true;
		switch (head->mode) {
			case RPC_REQ: {
				Packet packet(std::move(*buffer));
				if (packet.valid()) {
					onRequest(packet);
					pSession->add_sndmsg(std::make_shared<KRPC::Buffer>(packet.release()));
				} else {
					ret = false;	//对于不合理的请求立即关闭连接
				}
				break;
			}
			case RPC_SUB: {
				pSession->keep(true);	//对于订阅的session需要保活
				pSession->userdata.id = head->id;
				pSession->userdata.tp = std::chrono::steady_clock::time_point::min();
				auto it = subs[head->name].emplace(pSession);
				if (it.second && onSubcribe)
					onSubcribe(head->name);
				break;
			}
			case RPC_HEARTBEAT: {
				pSession->add_sndmsg(std::move(buffer));
				break;
			}
			default: {
				ret = false;	//对于不合理的请求立即关闭连接
				break;
			}
		}
		return ret;
	}
	
#if 0	//for debug
	virtual void onSend(typename Base::SessionSPtr pSession) {std::cout << pSession->address() << " send over!" << std::endl;}
	virtual void onConnect(typename Base::SessionSPtr pSession) {std::cout << pSession->address() << " connected!" << std::endl;}
	virtual void onClose(const std::string &ip, typename Base::UserData &data) {std::cout << ip << " closed!" << std::endl;}
#endif
private:
	typename ServerKind<TYPE>::start_arg_type arg;
	const AccessCallback access;
	std::function<void(Packet &)> onRequest;
	SubcribeCallback onSubcribe;

	using SWPtr = typename Base::SessionWPtr;
	using SubSet = std::set<SWPtr, std::owner_less<SWPtr>>;
	std::unordered_map<std::string, SubSet> subs;
};
#ifdef __linux__
using LSS = NetServer<ServerType::LOCALSTREAM>;
using LDS = NetServer<ServerType::LOCALDGRAM>;
#endif
using TS = NetServer<ServerType::TCP>;
using US = NetServer<ServerType::UDP>;

}
